package com.pulse.mqtt;

import com.pulse.mqtt.config.MqttConfiguration;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;

import java.nio.charset.Charset;

/**
 * @ClassName MqttConsumer
 * @Description
 * @Author 陆帅
 * @Date 2024/7/15 13:38
 */
public class MqttConsumer {
    public final static Logger log = LoggerFactory.getLogger(MqttConsumer.class);

    private volatile static MqttConsumer singleton;

    private MqttConsumer(){}

    public static MqttConsumer getSingleton(){
        if(singleton == null){
            synchronized (MqttConsumer.class){
                if(singleton == null){
                    singleton = new MqttConsumer();
                }
            }
        }
        return singleton;
    }

    public static MqttClient client;

    public MqttConfiguration mqttConfiguration;

    public void setMqttConfiguration(String host, String client, String username, String password, String topic, String qos, int keepAlive, int timeout){
        mqttConfiguration = new MqttConfiguration();
        mqttConfiguration.setHost(host);
        mqttConfiguration.setClientId(client);
        mqttConfiguration.setUsername(username);
        mqttConfiguration.setPassword(password);
        mqttConfiguration.setTopic(topic);
        mqttConfiguration.setQos(qos);
        mqttConfiguration.setKeepalive(keepAlive);
        mqttConfiguration.setTimeout(timeout);
    }

    public MqttConfiguration getMqttConfiguration(){
        return this.mqttConfiguration;
    }

    public void mqttRun(ApplicationArguments args){
        log.info("初始化并启动init start mqtt......");
        this.connect();
    }

    private void connect(){
        try{
            getClient();
        }
        catch (Exception e){
            log.error("mqtt连接异常: " + e);
        }
    }

    /**
     * 创建客户端
     */
    public void getClient(){
        try{
            if(null == client){
                client = new MqttClient(mqttConfiguration.getHost(),mqttConfiguration.getClientId(),new MemoryPersistence());
            }
            log.info("--创建mqtt客户端,Id:" + mqttConfiguration.getClientId() + " 服务器: " + mqttConfiguration.getHost());
        }catch (Exception e){
            log.error("创建mqtt客户端异常: " + e);
        }
    }

    public MqttClient getMqttClient(){
        return client;
    }

    /**
     * 生成配置对象，用户名，密码等
     * @param cof
     * @return
     */
    public MqttConnectOptions getOptions(MqttConfiguration cof){
        MqttConnectOptions options = new MqttConnectOptions();

        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(true);
        options.setUserName(cof.getUsername());
        options.setPassword(cof.getPassword().toCharArray());
        // 设置超时时间
        options.setConnectionTimeout(cof.getTimeout());
        // 设置会话心跳时间
        options.setKeepAliveInterval(cof.getKeepalive());
        // 是否清除session
        options.setAutomaticReconnect(true);
        options.setMaxReconnectDelay(60);
        log.info("--生成mqtt配置对象");
        return options;
    }

    /**
     * qos --- 3 ---
     *
     * @param length
     * @param qosss
     * @return
     */
    public int[] getQos(int length, String qosss) {
        String[] qoss = qosss.split(",", length);
        int[] qos = new int[length];
        for (int i = 0; i < length; i++) {
            /**
             * MQTT协议中有三种消息发布服务质量:
             *
             * QOS0： “至多一次”，消息发布完全依赖底层 TCP/IP
             * 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。 QOS1：
             * “至少一次”，确保消息到达，但消息重复可能会发生。 QOS2：
             * “只有一次”，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果，资源开销大
             */
            qos[i] = Integer.parseInt(qoss[i]);
        }
        log.info("--设置消息发布质量");
        return qos;
    }

    public void create(Object callback){
        try{
            client.setCallback((MqttCallback) callback);
            log.info("--添加回调处理类");
            client.connect(getOptions(mqttConfiguration));
        }catch (Exception e){
            log.info("装载实例或订阅主题异常: " + e + e.getStackTrace());
        }
    }

    /**
     * 订阅某个主题
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos){
        try{
            log.info("topic:" + topic);
            client.subscribe(topic, qos);
        }catch (MqttException e){
            log.error("异常: " + e);
        }
    }

    /**
     * 发布，非持久化
     * <p>
     * qos根据文档设置为1
     *
     * @param topic
     * @param msg
     */
    public static void publish(String topic, String msg) {
        publish(1, false, topic, msg);
    }

    /**
     * 发布，非持久化
     * <p>
     * qos根据文档设置为1
     *
     * @param topic
     * @param msg
     */
    public static void publish(String topic, byte[] msg) {
        publish(1, false, topic, msg);
    }

    /**
     * 发布
     *
     * @param qos
     * @param retained
     * @param topic
     * @param pushMessage
     */
    public static void publish(int qos, boolean retained, String topic, String pushMessage) {
        publish(qos, retained, topic, pushMessage.getBytes(Charset.forName("utf-8")));
    }

    public static void publish(int qos, boolean retained, String topic, byte[] pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage);
        MqttTopic mTopic = client.getTopic(topic);
        if (client == null) {
            log.error("client不存在");
            return;
        }
        if (null == mTopic) {
            log.error("topic:" + topic + " 不存在");
            return;
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();

            if (!token.isComplete()) {
                log.info("消息发送成功");
            }
        } catch (MqttPersistenceException e) {
            log.error("异常。", e);
        } catch (MqttException e) {
            log.error("异常。", e);
        }
    }
}
